package com.gzx.qc.modular.asr.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.ReUtil;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.google.common.collect.Lists;
import com.gzx.qc.core.config.QaKeywordInitializationConfig;
import com.gzx.qc.modular.asr.entity.GzxAsrTask;
import com.gzx.qc.modular.asr.entity.GzxAsrText;
import com.gzx.qc.modular.asr.service.GzxAsrDbService;
import com.gzx.qc.modular.asr.service.GzxAsrService;
import com.gzx.qc.modular.asr.strategy.baidu.dto.BaiduAsrTextRespDto;
import com.gzx.qc.modular.asr.strategy.baidu.dto.BaiduChatRespDto;
import com.gzx.qc.modular.asr.strategy.baidu.enums.BaiduTaskStatusEnum;
import com.gzx.qc.modular.asr.strategy.baidu.enums.BaiduTaskTypeEnum;
import com.gzx.qc.modular.asr.strategy.baidu.service.GzxBaiduService;
import com.gzx.qc.modular.asr.vo.GzxAsrTextVo;
import com.gzx.qc.modular.check.entity.GzxCallCheck;
import com.gzx.qc.modular.check.service.GzxCallCheckDbService;
import com.hankcs.hanlp.HanLP;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import vip.xiaonuo.common.consts.IntConstant;
import vip.xiaonuo.common.exception.CommonException;
import vip.xiaonuo.common.tenant.TenantContext;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author zjt
 * @description
 * @date 2025/7/29 15:34
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class GzxAsrServiceImpl implements GzxAsrService {

    private final GzxAsrDbService asrDbService;

    private final GzxBaiduService baiduService;

    private final GzxCallCheckDbService callCheckDbService;

    private final QaKeywordInitializationConfig qaKeywordInitializationConfig;

    /**
     * ASR转文本流程
     */
    @Override
    public void asrTextProcess() {
        List<GzxAsrTask> asrTaskList = asrDbService.selectTaskList(Wrappers.<GzxAsrTask>lambdaQuery()
                .select(GzxAsrTask::getId, GzxAsrTask::getFileUrl).isNull(GzxAsrTask::getTaskStatus).last("limit 100"));
        if (CollectionUtil.isNotEmpty(asrTaskList)) {
            for (GzxAsrTask asrTask : asrTaskList) {
                String taskId = baiduService.createTask(BaiduTaskTypeEnum.CREATE, asrTask.getFileUrl());
                asrTask.setTaskStatus(BaiduTaskStatusEnum.FAILURE.getStatus());
                if (StringUtils.isNotBlank(taskId)) {
                    asrTask.setTaskId(taskId);
                    asrTask.setTaskStatus(BaiduTaskStatusEnum.RUNNING.getStatus());
                }
                asrDbService.updateTaskById(asrTask);
            }
        }
    }

    /**
     * 同步百度云转写结果
     */
    @Override
    public void syncAsrText() {
        List<GzxAsrTask> asrTaskList = asrDbService.selectTaskList(Wrappers.<GzxAsrTask>lambdaQuery()
                .select(GzxAsrTask::getId, GzxAsrTask::getCallId, GzxAsrTask::getTaskId)
                .eq(GzxAsrTask::getTaskStatus, BaiduTaskStatusEnum.RUNNING.getStatus())
                .last("limit 100"));
        if (CollectionUtil.isNotEmpty(asrTaskList)) {
            List<String> taskIdList = asrTaskList.stream().map(GzxAsrTask::getTaskId).toList();
            List<BaiduAsrTextRespDto.BaiduAsrTaskInfo> baiduAsrTaskInfoList = baiduService.syncTaskResult(BaiduTaskTypeEnum.QUERY, taskIdList);
            if (CollectionUtil.isNotEmpty(baiduAsrTaskInfoList)) {
                Map<String, List<BaiduAsrTextRespDto.BaiduAsrTaskInfo>> asrTaskStatusMap = baiduAsrTaskInfoList.stream()
                        .collect(Collectors.groupingBy(BaiduAsrTextRespDto.BaiduAsrTaskInfo::getTask_status));
                for (String key : asrTaskStatusMap.keySet()) {
                    List<BaiduAsrTextRespDto.BaiduAsrTaskInfo> asrTaskInfoList = asrTaskStatusMap.get(key);
                    //修改主表状态
                    if (BaiduTaskStatusEnum.SUCCESS.getStatus().equals(key)) {
                        List<String> successTaskIdList = asrTaskInfoList.stream().map(BaiduAsrTextRespDto.BaiduAsrTaskInfo::getTask_id).toList();
                        asrDbService.updateTask(Wrappers.<GzxAsrTask>lambdaUpdate()
                                .set(GzxAsrTask::getTaskStatus, BaiduTaskStatusEnum.SUCCESS.getStatus())
                                .in(GzxAsrTask::getTaskId, successTaskIdList));
                        Map<String, String> taskIdMap = asrTaskList.stream().collect(Collectors.toMap(GzxAsrTask::getTaskId, GzxAsrTask::getCallId, (oldValue, newValue) -> newValue));
                        List<GzxAsrText> asrTextList = new ArrayList<>();
                        asrTaskInfoList.forEach(s -> asrTextList.addAll(convert(s, taskIdMap.get(s.getTask_id()))));
                        //保存文本
                        batchSaveText(asrTextList);
                        //补全关键词
                        completionCheckKeyword(asrTextList);
                    }
                    if (BaiduTaskStatusEnum.FAILURE.getStatus().equals(key)) {
                        List<String> failureTaskIdList = asrTaskStatusMap.get(key).stream().map(BaiduAsrTextRespDto.BaiduAsrTaskInfo::getTask_id).toList();
                        asrDbService.updateTask(Wrappers.<GzxAsrTask>lambdaUpdate()
                                .set(GzxAsrTask::getTaskStatus, BaiduTaskStatusEnum.FAILURE.getStatus())
                                .in(GzxAsrTask::getTaskId, failureTaskIdList));
                        //获取语音文字失败，提前进行AI审核，设置为双方均无异常
                        inAdvanceAiReview(asrTaskList.stream().map(GzxAsrTask::getCallId).filter(failureTaskIdList::contains).toList());
                    }
                }
            }
        }
    }

    /**
     * 分析语音文本
     */
    @Override
    public void unparkVoiceText() {
        //查询待AI审核的CallId
        List<GzxCallCheck> callCheckList = callCheckDbService.selectList(Wrappers.<GzxCallCheck>lambdaQuery()
                .select(GzxCallCheck::getId, GzxCallCheck::getCallId)
                .eq(GzxCallCheck::getAiReviewStatus, 0)
                .last("limit 100"));
        if (CollectionUtil.isNotEmpty(callCheckList)) {
            List<String> callIdList = callCheckList.stream().map(GzxCallCheck::getCallId).toList();
            List<GzxAsrTextVo> asrTextVoList = asrDbService.selectVoAsrText(Wrappers.<GzxAsrText>lambdaQuery()
                    .select(GzxAsrText::getCallId, GzxAsrText::getSpeaker, GzxAsrText::getContent, GzxAsrText::getBeginTime)
                    .in(GzxAsrText::getCallId, callIdList));
            if (CollectionUtil.isNotEmpty(asrTextVoList)) {
                Map<String, GzxCallCheck> callIdMap = callCheckList.stream()
                        .collect(Collectors.toMap(
                                GzxCallCheck::getCallId,
                                Function.identity(),
                                (existing, replacement) -> existing));
                Map<String, List<GzxAsrTextVo>> asrTextVoMap = asrTextVoList.stream().collect(Collectors.groupingBy(GzxAsrTextVo::getCallId));
                List<Future<GzxCallCheck>> futureList = new ArrayList<>();
                try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
                    for (Map.Entry<String, List<GzxAsrTextVo>> entry : asrTextVoMap.entrySet()) {
                        final String callId = entry.getKey();
                        futureList.add(executor.submit(() -> sendHttpAndReturn(entry.getValue(), callIdMap.get(callId))));
                    }
                }
                //等待所有线程执行完毕
                List<GzxCallCheck> resultCallCheckList = waitThread(futureList, callIdList);
                if (CollectionUtil.isNotEmpty(resultCallCheckList)) {
                    callCheckDbService.modifyAiReviewStatus(resultCallCheckList);
                }
            }
        }
    }

    /**
     * 查询ASR文字
     *
     * @param callId
     * @return
     */
    @Override
    public List<GzxAsrTextVo> getAstText(String callId) {
        List<GzxAsrTextVo> asrTextVoList = asrDbService.selectVoAsrText(Wrappers.<GzxAsrText>lambdaQuery()
                .select(GzxAsrText::getSpeaker, GzxAsrText::getContent, GzxAsrText::getBeginTime)
                .eq(GzxAsrText::getCallId, callId));
        asrTextVoList.sort((o1, o2) -> o1.getBeginTime() - o2.getBeginTime());
        return asrTextVoList;
    }

    private void batchSaveText(List<GzxAsrText> asrTextList) {
        Thread.startVirtualThread(() -> TenantContext.ignoreHandle(() -> {
            List<List<GzxAsrText>> batchList = Lists.partition(asrTextList, IntConstant.VALUE_1000);
            batchList.forEach(asrDbService::batchSaveText);
        }));
    }

    private void completionCheckKeyword(List<GzxAsrText> asrTextList) {
        Thread.startVirtualThread(() -> {
            List<GzxCallCheck> callCheckList = filterKeyword(asrTextList);
            if (CollectionUtil.isNotEmpty(callCheckList)) {
                List<List<GzxCallCheck>> batchList = Lists.partition(callCheckList, IntConstant.VALUE_500);
                TenantContext.ignoreHandle(() -> batchList.forEach(callCheckDbService::completionCheckKeyword));
                TenantContext.ignoreHandle(() -> inAdvanceAiReview(callCheckList.stream()
                        .filter(item -> item != null
                                && item.getAiReviewStatus() != null
                                && item.getAiReviewStatus() == 4
                                && item.getCallId() != null).map(GzxCallCheck::getCallId).toList()));
            }
        });
    }

    private void inAdvanceAiReview(List<String> callIdList) {
        if (CollectionUtil.isNotEmpty(callIdList)) {
            callCheckDbService.updateByWrapper(Wrappers.<GzxCallCheck>lambdaUpdate()
                    .set(GzxCallCheck::getAiReviewStatus, 4)
                    .set(GzxCallCheck::getAiReviewResult, "录音转写失败")
                    .in(GzxCallCheck::getCallId, callIdList));
        }
    }

    private List<GzxCallCheck> filterKeyword(List<GzxAsrText> asrTextList) {
        List<GzxCallCheck> callCheckList = new ArrayList<>();
        Map<String, List<GzxAsrText>> asrTextMap = asrTextList.stream().collect(Collectors.groupingBy(GzxAsrText::getCallId));
        for (String callId : asrTextMap.keySet()) {
            List<GzxAsrText> callAsrTextList = asrTextMap.get(callId);
            Set<String> keywordResult = new HashSet<>();
            for (GzxAsrText asrText : callAsrTextList) {
                Set<String> contentKeySet = HanLP.segment(asrText.getContent()).stream()
                        .map(term -> term.word)
                        .collect(Collectors.toCollection(HashSet::new));
                Set<String> keywordSet = qaKeywordInitializationConfig.verifyKeyWord(asrText.getSpeaker(), contentKeySet);
                if (CollectionUtil.isNotEmpty(keywordSet)) {
                    keywordResult.addAll(keywordSet);
                }
            }
            //是否有客户说的话
            if (asrTextList.stream().noneMatch(item -> item.getSpeaker() == 1)) {
                //不存在客户说的话
                keywordResult.add("空挂通时");
            }
            //录音为空
            if (CollectionUtil.isEmpty(callAsrTextList)) {
                GzxCallCheck callCheck = new GzxCallCheck();
                callCheck.setCallId(callId);
                callCheck.setAiReviewStatus(4);
                callCheck.setAiReviewResult("录音转写失败");
                callCheckList.add(callCheck);
            }
            if (CollectionUtil.isNotEmpty(keywordResult)) {
                GzxCallCheck callCheck = new GzxCallCheck();
                callCheck.setCallId(callId);
                callCheck.setKeyword(String.join(",", keywordResult));
                callCheckList.add(callCheck);
            }

        }
        return callCheckList;
    }

    private List<GzxAsrText> convert(BaiduAsrTextRespDto.BaiduAsrTaskInfo taskInfo, String callId) {
        List<GzxAsrText> result = new ArrayList<>();
        // 处理左声道
        if (taskInfo.getTask_result().getLeft() != null) {
            processChannel(result, taskInfo.getTask_result().getLeft().getDetailed_result(), 1, taskInfo.getTask_id(), callId);
        }
        // 处理右声道
        if (taskInfo.getTask_result().getRight() != null) {
            processChannel(result, taskInfo.getTask_result().getRight().getDetailed_result(), 2, taskInfo.getTask_id(), callId);
        }
        return result;
    }

    private void processChannel(List<GzxAsrText> resultList, List<BaiduAsrTextRespDto.SegmentResult> segments, int speaker, String taskId, String callId) {
        if (CollectionUtil.isNotEmpty(segments)) {
            for (BaiduAsrTextRespDto.SegmentResult segment : segments) {
                if (CollectionUtil.isNotEmpty(segment.getRes())) {
                    GzxAsrText text = new GzxAsrText();
                    text.setCallId(callId);
                    text.setTaskId(taskId);
                    text.setSpeaker(speaker);
                    text.setBeginTime(segment.getBegin_time());
                    text.setContent(String.join("", segment.getRes()));
                    text.setAllDetail(JSON.toJSONString(segment));
                    resultList.add(text);
                }
            }
        }
    }

    private GzxCallCheck sendHttpAndReturn(List<GzxAsrTextVo> asrTextVoMapList, GzxCallCheck callCheck) {
        asrTextVoMapList.sort((o1, o2) -> o1.getBeginTime() - o2.getBeginTime());
        StringBuilder sb = new StringBuilder("请告诉我，以下这段对话在催收场景下是否违规，比如坐席态度不好，骂人，嘲讽或者阴阳客户或者客户有投诉、情绪激动、自杀等危险信号，或者坐席有假冒身份，如法院或者公检法等的问题，但是使用调解中心的身份的话就是合法，你只需要回复1，2，3，4。1代表客户违规，2代表坐席违规，3代表双方都违规，4代表双方都不违规。强调一下你只需要回复一个数字就行\n");
        for (GzxAsrTextVo asrTextVo : asrTextVoMapList) {
            sb.append(asrTextVo.getSpeaker() == 1 ? "客户说：" : "坐席说：").append(asrTextVo.getContent()).append("\n");
        }
        try {
            BaiduChatRespDto.Choice.Message message = baiduService.unparkVoiceText(BaiduTaskTypeEnum.CHAT, sb.toString());
            if (message != null) {
                callCheck.setAiReviewStatus(Integer.valueOf(ReUtil.get("\\d+", message.getContent(), 0)));
                callCheck.setAiReviewResult(message.getReasoning_content());
            }
        } catch (Exception e) {
            callCheck.setAiReviewStatus(5);
            callCheck.setAiReviewResult(e.getMessage());
        }
        return callCheck;
    }

    private List<GzxCallCheck> waitThread(List<Future<GzxCallCheck>> futureList, List<String> callIdList) {
        return futureList.stream()
                .map(f -> {
                    String callId = futureList.indexOf(f) < callIdList.size() ? callIdList.get(futureList.indexOf(f)) : "UNKNOWN";
                    try {
                        return f.get(50, TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        throw new CommonException("CallId：{}执行AI质检任务线程中断", callId, e);
                    } catch (ExecutionException e) {
                        log.error("CallId：{}执行AI质检任务出现异常", callId, e);
                        return null;
                    } catch (TimeoutException e) {
                        log.warn("CallId：{}执行AI质检任务请求超时", callId);
                        return null;
                    }
                })
                .filter(Objects::nonNull)
                .toList();
    }

    public static void main(String[] args) {

    }
}
